-module (osmos_table_format).

-export ([ valid/1,
	   binary_replace/1,
	   term_replace/1,
	   term_sum_uint63_delete/1,
	   binary_sum_uint64_vector/1 ]).

-export ([ erlang_less/2 ]).

-ifdef (HAVE_EUNIT).
-include_lib ("eunit/include/eunit.hrl").
-endif.

-include ("osmos.hrl").

%
% public
%

valid (#osmos_table_format { block_size = BlockSize,
			     key_format = KeyFormat = #osmos_format{},
			     key_less = KeyLess,
			     value_format = ValueFormat = #osmos_format{},
			     merge = Merge,
			     short_circuit = ShortCircuit,
			     delete = Delete })
	when is_integer (BlockSize),
	     is_function (KeyFormat#osmos_format.to_binary, 1),
	     is_function (KeyFormat#osmos_format.from_binary, 1),
	     is_function (KeyLess, 2),
	     is_function (ValueFormat#osmos_format.to_binary, 1),
	     is_function (ValueFormat#osmos_format.from_binary, 1),
	     is_function (Merge, 3),
	     is_function (ShortCircuit, 2),
	     is_function (Delete, 2) ->
  true;
valid (_) ->
  false.

binary_replace (BlockSize) when is_integer (BlockSize),
				BlockSize >= ?OSMOS_MIN_BLOCK_SIZE ->
  BinaryFormat = osmos_format:binary (),
  ErlangLess = fun ?MODULE:erlang_less/2,
  Merge = fun merge_return_later/3,
  ShortCircuitAlways = fun short_circuit_always/2,
  DeleteNever = fun delete_never/2,
  #osmos_table_format { block_size = BlockSize,
			key_format = BinaryFormat,
			key_less = ErlangLess,
			value_format = BinaryFormat,
			merge = Merge,
			short_circuit = ShortCircuitAlways,
			delete = DeleteNever }.

term_replace (BlockSize) when is_integer (BlockSize),
			      BlockSize >= ?OSMOS_MIN_BLOCK_SIZE ->
  TermFormat = osmos_format:term (),
  ErlangLess = fun ?MODULE:erlang_less/2,
  Merge = fun merge_return_later/3,
  ShortCircuitAlways = fun short_circuit_always/2,
  DeleteNever = fun delete_never/2,
  #osmos_table_format { block_size = BlockSize,
			key_format = TermFormat,
			key_less = ErlangLess,
			value_format = TermFormat,
			merge = Merge,
			short_circuit = ShortCircuitAlways,
			delete = DeleteNever }.

term_sum_uint63_delete (BlockSize)
	when is_integer (BlockSize), BlockSize >= ?OSMOS_MIN_BLOCK_SIZE ->
  KeyFormat = osmos_format:term (),
  ErlangLess = fun ?MODULE:erlang_less/2,
  ValueToBinary = fun
		    (N) when is_integer (N), N >= 0, N < 16#7000000000000000 ->
		      <<0:1, N:63>>;
		    (delete) ->
		      <<>>
		  end,
  ValueFromBinary = fun
		      (<<_:1, N:63>>) -> N;
		      (<<>>) -> delete
		    end,
  ValueFormat = #osmos_format { to_binary = ValueToBinary,
				from_binary = ValueFromBinary },
  Merge = fun
	    (_K, _A, delete) -> delete;
	    (_K, delete, B) when is_integer (B) -> B;
	    (_K, A, B) when is_integer (A), is_integer (B) -> A + B
	  end,
  ShortCircuit = fun
		   (_K, delete) -> true;
		   (_K, N) when is_integer (N) -> false
		 end,
  Delete = fun (_Key, Value) -> Value =:= delete end,
  #osmos_table_format { block_size = BlockSize,
			key_format = KeyFormat,
			key_less = ErlangLess,
			value_format = ValueFormat,
			merge = Merge,
			short_circuit = ShortCircuit,
			delete = Delete }.

binary_sum_uint64_vector (BlockSize)
	when is_integer (BlockSize), BlockSize > ?OSMOS_MIN_BLOCK_SIZE ->
  KeyFormat = osmos_format:binary (),
  ErlangLess = fun ?MODULE:erlang_less/2,
  ValueFormat = osmos_format:uint64_vector (),
  MergeSum = fun merge_uint64_vector_sum/3,
  ShortCircuitNever = fun short_circuit_never/2,
  DeleteNever = fun delete_never/2,
  #osmos_table_format { block_size = BlockSize,
			key_format = KeyFormat,
			key_less = ErlangLess,
			value_format = ValueFormat,
			merge = MergeSum,
			short_circuit = ShortCircuitNever,
			delete = DeleteNever }.

%
% private
%

%% @hidden
erlang_less (A, B) ->
  A < B.

merge_return_later (_Key, _EarlierValue, LaterValue) ->
  LaterValue.

short_circuit_always (_Key, _Value) ->
  true.

short_circuit_never (_Key, _Value) ->
  false.

delete_never (_Key, _Value) ->
  false.

merge_uint64_vector_sum (_Key, V1, V2) ->
  uint64_vector_sum (tuple_to_list (V1), tuple_to_list (V2), []).

uint64_vector_sum ([ C1 | C1s ], [ C2 | C2s ], Acc) ->
  uint64_vector_sum (C1s, C2s, [ C1 + C2 | Acc ]);
uint64_vector_sum (C1s, [], Acc) when C1s =/= [] ->
  list_to_tuple (lists:reverse (lists:reverse (C1s) ++ Acc));
uint64_vector_sum ([], C2s, Acc) when C2s =/= [] ->
  list_to_tuple (lists:reverse (lists:reverse (C2s) ++ Acc));
uint64_vector_sum ([], [], Acc) ->
  list_to_tuple (lists:reverse (Acc)).

%
% tests
%

-ifdef (EUNIT).

basic_test () ->
  ok.

-endif.
